package com.beiding.dbrequest.common;

import com.beiding.dbrequest.part3.LastElementSubscriber;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.reactivex.Observable;
import lombok.extern.java.Log;

import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 业务行为分发器
 */
@Log
public class Dispatcher {

    /**
     * 消息产生时处理逻辑
     *
     * @param ctx
     * @param req
     * @throws Exception
     */
    public void onMessage(ChannelHandlerContext ctx, Message req) throws Exception {
        Message invoke = invoke(ctx, req);
        if (invoke instanceof ObservableMessage) {
            Observable<?> observable = ((ObservableMessage) invoke).getObservable();

            observable.subscribe(new LastElementSubscriber<Object>() {

                @Override
                public void onError(Throwable e) {
                    ctx.writeAndFlush(Message.responseFor(req).status(Message.STATUS_ERROR).msg(e.getMessage())).addListener((ChannelFutureListener) future -> {
                        if (!future.isSuccess()) {
                            future.cause().printStackTrace();
                        }
                    });
                }

                @Override
                public void gotLast(Object o) {
                    if (o instanceof Serializable||o==null) {
                        ctx.writeAndFlush(Message.responseFor(req).status(Message.STATUS_SUCCESS).body((Serializable)o)).addListener((ChannelFutureListener) future -> {
                            if (!future.isSuccess()) {
                                future.cause().printStackTrace();
                            }
                        });
                    }else{
                        log.warning("不支持的数据格式");
                    }
                }
            });

        } else {
            ctx.writeAndFlush(invoke).addListener((ChannelFutureListener) future -> {
                if (!future.isSuccess()) {
                    future.cause().printStackTrace();
                }
            });
        }

    }

    /**
     * 调用control处理请求头
     *
     * @param ctx
     * @param req
     * @return
     */
    private Message invoke(ChannelHandlerContext ctx, Message req) {
        String uri = req.getUri();
        Control control = controlMap.get(uri);
        if (control == null) {
            return Message.responseFor(req)
                    .status(Message.STATUS_NOT_FOUND)
                    .msg("不支持处理该uri: " + uri);
        } else {
            try {
                Request request = new Request();
                //设置上下文
                RequestContext.CURRENT_REQUEST.set(request);
                request.setId(req.getRequestId());
                request.setCtx(ctx);
                request.setRequestMessage(req);
                request.setGlobal(global);

                //返回结果可能是promise
                Object result = control.call(request);
                if (result == null || result instanceof Serializable) {
                    Serializable r = (Serializable) result;
                    return Message.responseFor(req)
                            .body(r);
                } else if (result instanceof Observable) {
                    return new ObservableMessage((Observable) result);
                } else {
                    throw new RuntimeException("不支持的响应类型:" + result.getClass());
                }
            } catch (InvocationTargetException e) {
                e.printStackTrace();
                return Message.responseFor(req)
                        .status(Message.STATUS_ERROR)
                        .msg(e.getTargetException().getMessage());
            } catch (Exception e) {
                e.printStackTrace();
                return Message.responseFor(req)
                        .status(Message.STATUS_ERROR)
                        .msg(e.getMessage());
            } finally {
                //清理上下文
                RequestContext.CURRENT_REQUEST.remove();
            }
        }

    }

    public void addController(Object first, Object... controllers) {
        for (Object controller : Arrays.asList(first, controllers)) {
            Class<?> type = controller.getClass();
            Mapping annotation = type.getAnnotation(Mapping.class);
            String base = "";
            if (annotation != null) {
                base = annotation.value();
            }
            for (Method method : type.getMethods()) {
                Mapping mm = method.getAnnotation(Mapping.class);
                method.setAccessible(true);
                if (mm != null) {
                    Class<?>[] parameterTypes = method.getParameterTypes();
                    if (parameterTypes.length != 1 && parameterTypes[0] != Request.class) {
                        throw new RuntimeException("方法参数不符合要求: " + method);
                    }
                    String uri = base + mm.value();
                    Control control = addControl(uri, new Control() {
                        @Override
                        public Object call(Request request) throws Exception {
                            return method.invoke(controller, request);
                        }

                        @Override
                        public Object controller() {
                            return controller;
                        }
                    });
                    if (control != null) {
                        log.info("映射方法覆盖: " + uri);
                    }
                }
            }
        }

    }

    /**
     * 使用一个Map批量添加control
     *
     * @param controlMap
     * @return 被替换的control
     */
    public Map<String, Control> addControl(Map<String, Control> controlMap) {

        Map<String, Control> rp = new HashMap<>();
        controlMap.forEach((k, v) -> {
            rp.put(k, addControl(k, v));
        });
        return rp;
    }

    public Control addControl(String uri, Control Control) {
        return controlMap.put(uri, Control);
    }

    public void removeController(Object controller) {
        Iterator<Map.Entry<String, Control>> iterator = controlMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, Control> next = iterator.next();
            Control value = next.getValue();
            if (value.controller() == controller) {
                iterator.remove();
            }
        }
    }

    public Control getControl(String uri) {
        return controlMap.get(uri);
    }

    public Control removeControl(String uri) {
        return controlMap.remove(uri);
    }

    public Set<Object> getAllController() {
        Set<Object> r = new HashSet<>();
        for (Control value : controlMap.values()) {
            Object controller = value.controller();
            if (controller != null) {
                r.add(controller);
            }
        }
        return r;
    }

    public void intercept(String uri, Intercept intercept) {
        Control control = getControl(uri);
        if (control == null) {
            throw new IllegalArgumentException("未找到控制器:" + uri);
        }
        intercept.setTarget(control);
        addControl(uri, intercept);
    }

    //全局变量
    private Map<String, Object> global = new ConcurrentHashMap<>();

    //控制器map
    private Map<String, Control> controlMap = new ConcurrentHashMap<>();

    public Map<String, Object> getGlobal() {
        return global;
    }

    /**
     * 调用后忽略结果
     * 可用于广播
     *
     * @param ctx
     * @param req
     */
    public void onMessageIgnoreResult(ChannelHandlerContext ctx, Message req) {
        //调用完成后不需要返回结果
        invoke(ctx, req);
    }

    public Map<String, Control> getControlMap() {
        return controlMap;
    }
}
